package com.atguigu.gmall.realtime.utils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.IOUtils;
import redis.clients.jedis.Jedis;

import java.util.List;
import java.util.Locale;

/**
 * 查询维度数据的工具类
 *
 * @author lvbingbing
 * @date 2022-04-07 20:19
 */
@Slf4j
public class DimUtils {

    /**
     * 使用旁路缓存查询维度数据。如果缓存中有，直接从缓存中取；缓存中没有，再从 phoenix 查询
     * key:  dim:表名:主键值1_主键值2；value：jsonString
     * <p>
     * ttl:1天
     *
     * @param tableName 表名
     * @param id        维度数据的主键
     * @return 满足条件的维度数据
     */
    public static JSONObject getDimInfo(String tableName, String id) {
        return getDimInfo(tableName, Tuple2.of("id", id));
    }

    /**
     * 使用旁路缓存查询维度数据。如果缓存中有，直接从缓存中取；缓存中没有，再从 phoenix 查询
     * <p>
     * key:  dim:表名:主键值1_主键值2；value：jsonString
     * <p>
     * ttl:1天
     *
     * @param tableName       表名
     * @param colNameAndValue 列名和列值
     * @return 满足条件的维度数据
     */
    public static JSONObject getDimInfo(String tableName, Tuple2<String, String>... colNameAndValue) {
        // 1、拼接 redis key
        StringBuilder redisKeyBuilder = new StringBuilder();
        redisKeyBuilder.append("dim:")
                .append(tableName.toLowerCase())
                .append(":");
        for (Tuple2<String, String> tuple2 : colNameAndValue) {
            String columnValue = tuple2.f1;
            redisKeyBuilder.append(columnValue)
                    .append("_");
        }
        String redisKey = redisKeyBuilder.substring(0, redisKeyBuilder.length() - 1);
        Jedis jedis = null;
        try {
            jedis = RedisUtils.getJedis();
            String jsonString = jedis.get(redisKey);
            if (StringUtils.isEmpty(jsonString)) {
                // 缓存中没有数据，直接从 phoenix 查询
                log.info("缓存中没有 redisKey：{} 对应的数据，直接从 phoenix 查询！", redisKey);
                JSONObject jsonObject = getDimInfoNoCache(tableName, colNameAndValue);
                // 从 phoenix 查询到数据时才往 redis 中放数据
                if (MapUtils.isNotEmpty(jsonObject)) {
                    // 将查询到的数据放入缓存中
                    jedis.setex(redisKey, 60 * 60 * 24, jsonObject.toJSONString());
                    log.info("将查询到的维度数据放入缓存中！redisKey：{}", redisKey);
                }
                return jsonObject;
            }
            // 缓存中有数据，将其解析成 JSONObject 并返回
            return JSON.parseObject(jsonString);
        } catch (Exception e) {
            log.error("从 redis 中获取维度数据出现异常！为了保证程序正常执行，使用 phoenix 查询！redisKey：{}, 异常原因：{}", redisKey, e);
            return getDimInfoNoCache(tableName, colNameAndValue);
        } finally {
            IOUtils.closeQuietly(jedis);
        }
    }

    /**
     * 直接从 phoenix 查询，没有缓存
     *
     * @param tableName       表名
     * @param colNameAndValue 列名和列值
     * @return 满足条件的维度数据
     */
    public static JSONObject getDimInfoNoCache(String tableName, Tuple2<String, String>... colNameAndValue) {
        // 组装 where 条件
        StringBuilder whereSql = new StringBuilder(" where ");
        for (Tuple2<String, String> tuple : colNameAndValue) {
            String columnName = tuple.f0;
            String columnValue = tuple.f1;
            whereSql.append(columnName)
                    .append(" = '")
                    .append(columnValue)
                    .append("'")
                    .append(" and ");
        }
        String substring = whereSql.substring(0, whereSql.length() - 5);
        String sql = "select * from " + tableName + substring;
        log.info("查询维度 SQL：{}", sql);
        List<JSONObject> dimList = PhoenixUtils.queryList(sql, JSONObject.class);
        if (CollectionUtils.isEmpty(dimList)) {
            log.info("未查询到维度数据！");
            return new JSONObject(0);
        }
        // 因为关联维度，肯定都是根据key关联得到一条记录
        return dimList.get(0);
    }

    /**
     * 清空缓存
     *
     * @param tableName 表名
     * @param id        维度数据的主键
     */
    public static void clearCache(String tableName, String id) {
        try (Jedis jedis = RedisUtils.getJedis()) {
            String redisKey = "dim:" + tableName.toLowerCase() + ":" + id;
            jedis.del(redisKey);
        } catch (Exception e) {
            log.error("删除缓存失败！待删除的缓存信息对应的表名：{}, 主键：{}, 异常原因：{}", tableName, id, e);
        }
    }

    public static void main(String[] args) {
        long startTime = System.currentTimeMillis();
        // JSONObject jsonObject = getDimInfoNoCache("dim_base_trademark", Tuple2.of("id", "16"), Tuple2.of("TM_NAME", "网易"));
        JSONObject jsonObject = getDimInfo("dim_base_trademark", "19");
        long endTime = System.currentTimeMillis();
        long costTime = endTime - startTime;
        log.info("查询耗时：{}ms, 查询到的维度数据：{}", costTime, jsonObject);
    }
}